use std::any::Any;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::SystemTime;
use bytes::Bytes;
use log::info;
use tokio::sync::mpsc;
use tokio::sync::oneshot::Sender;
use rpc::connection_manager::event::{ConnectionEvent, ConnectionManageable, ConnectionManagerDataEvent, IdleData, IdleListener, IdleResultData, ReadData, ResponseData, RpcIdleEvent, WriteData};
use rpc::{ChannelId};
use rpc::sender::RpcSender;
use crate::connection_manager::message::ConnectionMessage;

pub mod message;
pub mod client_connection;

pub struct ClientConnectionManager {
    sender_by_connection_id: HashMap<ChannelId, mpsc::UnboundedSender<Bytes>>,
    response_data: HashMap<String, Sender<ResponseData>>,
    write_read_tricks: HashMap<ChannelId, (SystemTime, SystemTime)>,
}

impl ClientConnectionManager {
    pub fn new() -> ClientConnectionManager {
        ClientConnectionManager {
            sender_by_connection_id: HashMap::new(),
            response_data: HashMap::new(),
            write_read_tricks: HashMap::new(),
        }
    }

    pub fn add_response_data(&mut self, local_seq: String, sender: Sender<ResponseData>) {
        self.response_data.insert(local_seq, sender);
    }

    pub fn remove_response_data(&mut self, path: &str)-> Option<Sender<ResponseData>> {
        self.response_data.remove(path)
    }
}

impl ConnectionManageable for ClientConnectionManager {
    fn handle_event(&mut self, event: Box<dyn ConnectionEvent>) {
        let as_any : Box<dyn Any> = event;
        if as_any.is::<ConnectionMessage>() {
            let message = as_any.downcast::<ConnectionMessage>().unwrap();
            handle_connection_message(*message, self);
        } else if as_any.is::<ConnectionManagerDataEvent>() {
            let message = as_any.downcast::<ConnectionManagerDataEvent>().unwrap();
            handle_connection_manager_data_event(*message, self);
        }
    }
}

fn handle_connection_message(event: ConnectionMessage, manager: &mut ClientConnectionManager) {
    match event {
        ConnectionMessage::Disconnection(id) => {
            manager.sender_by_connection_id.remove(&id);
        }
        ConnectionMessage::Write(data) => {
            let WriteData{channel_id, local_seq, data, has_response, sender} = data;
            match sender {
                None => {
                    let bytes = Bytes::from(data);
                    manager.sender_by_connection_id.get(&channel_id).unwrap().send(bytes).unwrap();
                }
                Some(sender) => {
                    manager.add_response_data(local_seq, sender);
                    let bytes = Bytes::from(data);
                    manager.sender_by_connection_id.get(&channel_id).unwrap().send(bytes).unwrap();
                }
            }
        }
        ConnectionMessage::NewConnection { connection_id, sender, recv_sender: recev_sender } => {
            info!("handle client connection open: {:?}", connection_id);
            manager.sender_by_connection_id.insert(connection_id, sender.clone());
            recev_sender.send(true).ok();
        }
        ConnectionMessage::Response(data) => {
            let ResponseData{local_seq, is_success, data} = data;
            let sender = manager.remove_response_data(&local_seq);
            if let Some(sender) = sender {
                sender.send(ResponseData::new(local_seq, is_success, data)).unwrap();
            }
        }
    }
}

fn handle_connection_manager_data_event(event: ConnectionManagerDataEvent, manager: &mut ClientConnectionManager) {
    match event {
        ConnectionManagerDataEvent::Read(data) => {
            let ReadData{ channel_id, length} = data;
            println!("Received {} bytes from {:?}", length, channel_id);
            let time_option = manager.write_read_tricks.get_mut(&channel_id);
            if let Some((write_time, read_time)) = time_option {
                *read_time = SystemTime::now();
            } else {
                manager.write_read_tricks.insert(channel_id, (SystemTime::now(), SystemTime::now()));
            }
        }
        ConnectionManagerDataEvent::Idle{ write_idle_time, read_idle_time, sender, listeners } => {
            let datas = vec![IdleData::new(write_idle_time, read_idle_time)];
            handle_idle_data(datas, manager, sender, listeners);
        }
        ConnectionManagerDataEvent::Idles { data, sender, listeners } => {
            handle_idle_data(data, manager, sender, listeners);
        }
    }
}

fn handle_idle_data(data_list: Vec<IdleData>, manager: &mut ClientConnectionManager, mut sender: RpcSender, listeners: Arc<Vec<Box<dyn IdleListener>>>) {
    let now = SystemTime::now();
    let mut result = Vec::new();
    for (key, value) in manager.write_read_tricks.iter() {
        for data in data_list.iter() {
            let write_idle_time = now.duration_since(value.0).unwrap().as_secs();
            let read_idle_time = now.duration_since(value.1).unwrap().as_secs();

            let mut write_is_idle = false;
            if let Some(write_idle) = data.write_idle_time {
                if write_idle * 60 > write_idle_time {
                    write_is_idle = true;
                }
            }

            let mut read_is_idle = false;
            if let Some(read_idle) = data.read_idle_time {
                if read_idle * 60 > read_idle_time {
                    read_is_idle = true;
                }
            }

            if write_is_idle && read_is_idle {
                result.push(IdleResultData::new(key.clone(),
                                                Some(write_idle_time),
                                                Some(read_idle_time),
                ));
            }
        }
    }
    if result.len() > 0 {
        let event = RpcIdleEvent::new(result, listeners);
        let r = sender.send_event(Box::new(event));
    }
}